package com.fulihui.wgj.consumer.listener;

import com.fulihui.common.constants.TopicConstants;
import com.fulihui.wgj.consumer.communication.event.EventMsgProcessor;
import com.fulihui.wgj.consumer.communication.event.EventMsgProcessorRegister;
import com.fulihui.wgj.consumer.communication.normal.NormalMsgProcessor;
import com.fulihui.wgj.consumer.communication.normal.NormalMsgProcessorRegister;
import com.fulihui.wgj.consumer.service.Wx48HUserService;
import com.fulihui.wgj.facade.msg.EventMsg;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;


/**
 * 微信事件消息监听
 *
 * @author wang_wx
 * @date 2018-06-28
 */
@Component
@Slf4j
public class EventNotifyListener extends AbstractListener<EventMsg> {

    @Autowired
    private EventMsgProcessorRegister eventMsgProcessorRegister;
    @Autowired
    private NormalMsgProcessorRegister normalMsgProcessorRegister;
    @Autowired
    private Wx48HUserService wx48HUserService;
    @Autowired
    private ThreadPoolTaskExecutor threadPoolTaskExecutor;

    @Override
    @KafkaListener(topics = TopicConstants.EVENT_NOTIFY)
    public void listener(ConsumerRecord<String, String> record) {
        //1.解包 > 2.执行
        this.execute(unpack(record.value(), EventMsg.class));
    }

    @Override
    public void execute(EventMsg eventMsg) {

        if (eventMsg == null) {
            return;
        }
        // 48H 互动用户入库、更新
        threadPoolTaskExecutor.execute(()->{
            wx48HUserService.upsert48HUser(eventMsg);
        });

        log.info("MsgListener listener event = {},msgtype = {}", eventMsg.getEvent(),eventMsg.getMsgType());
        //  判断是否是 事件 通知
        EventMsgProcessor eventMsgProcessor = eventMsgProcessorRegister.getProcessor(eventMsg.getEvent());
        if(null != eventMsgProcessor){
            eventMsgProcessor.process(eventMsg);
            return;
        }

        // 判断是否是 用户消息 通知
        NormalMsgProcessor normalMsgProcessor = normalMsgProcessorRegister.getProcessor(eventMsg.getMsgType());
        if(null != normalMsgProcessor){
            normalMsgProcessor.process(eventMsg);
            return;
        }

        log.error("通知类型未定义={}",eventMsg);

    }


}
